package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2023/2/2
 *
 * 如果同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险
 *
 */
public class Demo13_LoginFail
{

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<LoginEvent> watermarkStrategy = WatermarkStrategy
            .<LoginEvent>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getEventTime());

        //读数据
        KeyedStream<LoginEvent, Long> ds = env
            .readTextFile("data/LoginLog.csv").setParallelism(1)
            .map(new MapFunction<String, LoginEvent>()
            {
                @Override
                public LoginEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    return new LoginEvent(
                        Long.parseLong(words[0]),
                        words[1],
                        words[2],
                        Long.parseLong(words[3]) * 1000
                    );
                }
            }).setParallelism(1)
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(LoginEvent::getUserId);

        /*
                定义一个模式，使用模式去匹配指定的数据
                        同一用户（可以是不同IP）在2秒之内连续两次登录，都失败

                        fail.next(fail)
         */
        Pattern<LoginEvent, LoginEvent> pattern = Pattern
            .<LoginEvent>begin("fail")
            .where(new SimpleCondition<LoginEvent>()
            {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equals(value.getEventType());
                }
            })
            //要求下一条必须在2s之内到达
            .within(Time.milliseconds(2001))
            .next("fail again")
            .where(new SimpleCondition<LoginEvent>()
            {
                @Override
                public boolean filter(LoginEvent value) throws Exception {
                    return "fail".equals(value.getEventType());
                }
            });

        PatternStream<LoginEvent> patternStream = CEP.pattern(ds, pattern);

        patternStream.select(new PatternSelectFunction<LoginEvent, String>()
        {
            /*
                    如果匹配成功:
                        {
                            fail=[ LoginEvent() ],
                            fail again = [ LoginEvent()]
                        }
             */
            @Override
            public String select(Map<String, List<LoginEvent>> pattern) throws Exception {
                LoginEvent event1 = pattern.get("fail").get(0);
                LoginEvent event2 = pattern.get("fail again").get(0);
                return event1.getUserId() + "在" + event1.getEventTime() +",和"+event2.getEventTime() +",恶意登录...";
            }
        })
                     .print();

        env.execute();
    }

}
